package com.sunrate.investingquote.service;

import cn.hutool.core.util.NumberUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.sunrate.investingquote.bean.Srcrate;
import com.sunrate.investingquote.client.CustomerWebSocketClient;
import com.sunrate.investingquote.dto.IvPairIdDTO;
import com.sunrate.investingquote.enums.RetCode;
import com.sunrate.investingquote.exception.ServerException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.regex.Matcher;
import java.util.stream.Collectors;

import static com.sunrate.investingquote.constant.Constants.*;

/**
 * @author yyh
 * @description:
 * @date 2023/06/15/ 15:10
 */
@Service
@Slf4j
public class QuoteWebSocketClientService {


    private static final BlockingQueue<String> QUOTE_QUEUE = new LinkedBlockingQueue<>(1000);


    @Value("${iv.pairids}")
    private String pairIds;
    @Value("${iv.srcid}")
    private String srcId;

    @Autowired
    @Qualifier("schedulerThreadPool")
    private ThreadPoolTaskScheduler schedulerThreadPool;
    @Autowired
    @Qualifier("serviceThreadPool")
    private ThreadPoolTaskExecutor serviceThreadPool;
    private final List<IvPairIdDTO> pairIdList = new ArrayList<>();


//    @PostConstruct
    public void init() {
        log.info("Init start...");
        loadIvPairds();
        consumerMsg();
        startClient();
        log.info("Init end...");
    }

    private void loadIvPairds() {
        if (StringUtils.isEmpty(srcId) || StringUtils.isEmpty(pairIds)) {
            return;
        }
        String[] split = pairIds.split(",");
        for (String s : split) {
            String[] ivArry = s.split(":");
            IvPairIdDTO iv = new IvPairIdDTO();
            iv.setPrdId(ivArry[0]);
            iv.setPairId(ivArry[1]);
            iv.setSrcId(srcId);
            pairIdList.add(iv);
        }
    }

    private void consumerMsg() {
        serviceThreadPool.execute(this::pushQuote);
    }

    private void pushQuote() {
        while (true) {
            try {
                log.info("pushQuote size:{},remain-size:{}", QUOTE_QUEUE.size(), QUOTE_QUEUE.remainingCapacity());
                Srcrate srcrate = parseToSrcrate(QUOTE_QUEUE.take());
                if (srcrate == null) {
                    continue;
                }
                log.info("send-------------- msg:{}", JSON.toJSONString(srcrate));
                // TODO http invoke

            } catch (InterruptedException e) {
                log.error("take msg InterruptedException queue size:{},remain-size:{}", QUOTE_QUEUE.size(), QUOTE_QUEUE.remainingCapacity(), e);
                Thread.currentThread().interrupt();
            } catch (Exception e) {
                log.error("pushQuote msg error queue size:{},remain-size:{}", QUOTE_QUEUE.size(), QUOTE_QUEUE.remainingCapacity(), e);
            }
        }
    }

    public void startClient() {
        create(WSS_URI);
    }

    public void startClient(String wss) {
        if (StringUtils.isEmpty(wss)) {
            log.warn("wss is isEmpty");
            wss = WSS_URI;
        }
        create(wss);
    }

    public void create(String wssUri) {
        CustomerWebSocketClient webSocketClient;
        try {
            webSocketClient = new CustomerWebSocketClient(wssUri);
        } catch (Exception e) {
            log.error("WebSocketClient create error:", e);
            throw new ServerException(RetCode.CREATE_CLIENT_ERROR);
        }
        webSocketClient.addOnCloseEventListener(obj -> create(WSS_URI));
        webSocketClient.addOnOpenEventListener(obj -> {
            //发送订阅的pid msg
            webSocketClient.send(subscribeMsg());
            //心跳
            schedulerThreadPool.schedule(() -> webSocketClient.send(HEART_BEAT_MSG), CRON_TRIGGER);
        });
        webSocketClient.addOnMessageEventListener(msg -> {
            if (msg == null || StringUtils.isEmpty((String) msg)) {
                log.warn("msg is empty msg:{}", msg);
                return;
            }
            if (!((String) msg).contains(MSG_MARK)) {
                return;
            }
            boolean offer = QUOTE_QUEUE.offer((String) msg);
            if (!offer) {
                log.info("offer msg fail msg:{}", msg);
            }
        });
        webSocketClient.connect();
        log.info("create end");
    }


    private String subscribeMsg() {
        Map<String, Object> map = new LinkedHashMap<>();
        map.put("_event", "bulk-subscribe");
        map.put("tzID", 8);
        String message;
        if (CollectionUtils.isEmpty(pairIdList)) {
            log.warn("ivPids is empty");
            message = StringUtils.EMPTY;
        } else {
            message = StringUtils.join(pairIdList.stream().map(ivPid -> "pid-" + ivPid.getPairId()).sorted().collect(Collectors.toList()), ":%%") + ":";
        }
        map.put("message", message);
        String s = String.format("[%s]", JSON.toJSONString(JSON.toJSONString(map)));
        log.info("subscribe pid:{}", s);
        return s;
    }


    private Srcrate parseToSrcrate(String msg) {
        try {
            //msg='a[{mssage:{}}]'
            JSONArray jsonArray = JSON.parseArray(msg.substring(1));
            JSONObject jsonObject = JSON.parseObject(jsonArray.getString(0));
            String message = jsonObject.getString(MSG_MARK);
            Matcher matcher = REG_PATTERN.matcher(message);
            if (!matcher.find()) {
                log.warn("message  is empty msg:{}", msg);
                return null;
            }
            JSONObject quote = JSON.parseObject(matcher.group());
            Optional<IvPairIdDTO> optional = pairIdList.stream().filter(o -> o.getPairId().equals(quote.getString("pid"))).findAny();
            if (!optional.isPresent()) {
                log.warn("ivPid is null msg:{}", msg);
                return null;
            }
            Srcrate srcrate = new Srcrate();
            srcrate.setPrdId(optional.get().getPrdId());
            srcrate.setSrcId(optional.get().getSrcId());
            BigDecimal bid = NumberUtil.toBigDecimal(quote.getString("bid"));
            BigDecimal ask = NumberUtil.toBigDecimal(quote.getString("ask"));
            BigDecimal spread = bid.subtract(ask).abs();
            BigDecimal mid = bid.add(ask).abs().divide(new BigDecimal("2"), 4, RoundingMode.HALF_UP);
            srcrate.setSrcAsk(ask);
            srcrate.setSrcBid(bid);
            srcrate.setSrcMid(mid);
            srcrate.setSrcSpread(spread);
            srcrate.setLastDatetime(LocalDateTime.now());
            srcrate.setSrcDatetime(LocalDateTime.ofInstant(Instant.ofEpochSecond(quote.getLong("timestamp")), ZoneId.systemDefault()));
            return srcrate;
        } catch (Exception e) {
            log.error("parse msg error msg:{}", msg, e);
        }
        log.warn("parse fail  msg:{}", msg);
        return null;
    }


}
